package com.macro.mall.demo.nio;


import java.io.IOException;
import java.nio.ByteBuffer;
import java.nio.channels.Pipe;
import java.util.concurrent.TimeUnit;

public class PipeDemo {
    public static void main(String[] args) throws IOException {
        Pipe pipe = Pipe.open();
        startProductThread(pipe);
        startConsumerThread(pipe);
    }

    public static void startProductThread(Pipe pipe) {
        Pipe.SinkChannel sink = pipe.sink(); // 输出通道
        ByteBuffer buffer = ByteBuffer.allocate(100);
        Runnable r = () -> {
            for (int i = 0; i < 3; i++) {
                try {
                    synchronized (PipeDemo.class) {
                        TimeUnit.SECONDS.sleep(1);
                        String message = String.format("[%s]%d time send message", Thread.currentThread().getName(), i);
                        buffer.put(message.getBytes());
                        buffer.flip(); // rest buffer
                        sink.write(buffer); //数据输出
                        buffer.clear();
                    }
                } catch (Exception e) {
                }
            }
        };
        for (int i = 0; i < 2; i++) {
            new Thread(r, "product-" + i).start();

        }
    }
    public static void startConsumerThread(Pipe pipe) {
        Pipe.SourceChannel source = pipe.source();
        ByteBuffer buffer = ByteBuffer.allocate(100);
        new Thread(() -> {
            while (true) {
                int len = 0;
                try {
                    while ((len = source.read(buffer)) != -1) {
                        buffer.flip(); // 切换读写
                        String r = new String(buffer.array(), 0, len);
                        System.out.println("r = " + r);
                        buffer.clear(); // 接受下次数据写入
                    }
                } catch (Exception e) {
                }
            }
        }).start();
    }

}

